package com.dongdongshop.mq;

import com.dongdongshop.service.AliyunCodeService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
@RocketMQMessageListener(topic = "my_code_topic",consumerGroup = "code_group")
public class getCodeMq implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private AliyunCodeService aliyunCodeService;

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void onMessage(String s) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                try{

                    for (MessageExt ext : list) {
                        byte[] body = ext.getBody();
                        String phone = new String(body);
                        if(!redisTemplate.hasKey(phone)){
                            String code = aliyunCodeService.aliyunCode(phone);
                            redisTemplate.boundValueOps(phone).set(code);
                            redisTemplate.expire(phone,30, TimeUnit.MINUTES);

                        }
                    }

                }catch (Exception e){
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
